package org.rency.mq.kafka.consumer.support.factory;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.core.ConsumerFactory;

import java.util.Map;

/**
 * 消费者工厂
 *
 * @author: Created by user_rcy@163.com on 2016/10/19.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class DefaultConsumerFactory<K, V> implements ConsumerFactory<K, V>{

    private Map<String, Object>                   consumerProperties;

    public DefaultConsumerFactory(Map<String, Object> consumerProperties){
        this.consumerProperties = consumerProperties;
    }

    @Override
    public Consumer<K, V> createConsumer() {
        return new KafkaConsumer<K, V>(this.consumerProperties);
    }

    @Override
    public boolean isAutoCommit() {
        Object auto = this.consumerProperties.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
        return auto instanceof Boolean ? (Boolean) auto : auto instanceof String ? Boolean.valueOf((String) auto) : false;
    }

}
